package com.dxf.bigdata.D04_spark

import org.apache.commons.logging.{Log, LogFactory}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 *  RDD方法主要有 转换 和 行动
 *   转换 如  Map , flatMap
 *   行动 collect
 *
 *
 *   RDD算子:   RDD的方法就是算子 --> 对数据的操作称为算子
 *
 *   RDD算子 => 转换算子 Map , 行动算子  collect
 *
 *   ===========================
 *
 *
 *   Map  => 处理数据,一个处理完了处理下一个 ==> 效率肯定底 (对比字节流) ==> 处理方式 缓存区
 *
 *
 *
 *  =====================================
 *
 *
 *   glom : 将同一个分区的数据直接转换为相同类型的内存数组进行处理, 分区不变
 *
 *    [1 ,2 ] [3 ,4 ] --> [List(1,2)] [List(3,4)]
 *
 *
 *   合并 RDD[Int]  -> RDD[Array[Int]]
 *
 *
 *   ===========================================
 *
 *
 *    groupBy 分组 ,同一组放在一起
 *    groupBy 会将数据打乱重新组合,我们将这样的操作称之为shuffle,极限情况下,数据可能被分再同一个分区中
 *
 *
 *    ===========================
 *
 *    Filter 过滤:
 *      将数据根据指定规则进行筛选过滤,符合规则的数据保留,不符合规则的数据丢弃
 *      当数据进行筛选后,分区不变,但是分区内的数据可能不均衡,生成环境下,可能会出现数据倾斜
 *
 *
 * =========================================
 *
 *
 * sample
 *   根据指定的规则,从数据集中抽取数据,
 *   使用场景:
 *    解决数据倾斜问题
 *
 *   def sample (
 *      withReplacement:Boolean,     =>  true 抽取后放回去,false 抽取后不放回去
 *      fraction:Double,             =>   比例
 *      seed:Long=Utils.random.nextLong
 *   ) : RDD[T]
 *
 *   =============================
 *
 *
 *   distinct 去重:
 *
 *   =========================
 *
 *   coalesce:
 *    根据数量缩减分区,用于大数据集过滤后,提高小数据集的执行效率
 *    当spark程序中,存在过多的小任务时候,可以通过coalesce方法,收缩合并分区,减少分区个数,减小任务调度成本
 *
 *
 *  =============================
 *
 *  sortBy 排序
 *
 *  ======================
 *
 *  def intersection(other:RDD[T]) : RDD[T]
 *  说明:
 *    对2个RDD求交集(intersection) 后返回一个新的RDD
 *
 *
 * def union(other:RDD[T]) : RDD[T] 并集
 *
 *
 * def subtract(other:RDD[T]) : RDD[T] 差集
 *
 * =============================================
 *
 *  键值操作: kv类型
 *
 *    partitionBy
 *
 *    ================================
 *
 *    reduceByKey 可以将数据按照相同的key对value进行聚合
 *
 *
 *
 *   ====================================
 *
 *
 *   groupByKey 将分区的数据直接转换为相同类型的内存数组,进行后续处理
 *
 *
 *   ================================
 *
 *   aggregateByKey: 将数据根据不同规则进行分区内计算和分区间计算
 *
 *    =========================================
 *
 *   combineByKey:
 *    和aggregateByKey类似,但是不需要初始值,把列表中的第一个元素作为初始值
 *     combineByKey:
 *     第一个参数 : 将相同的key的第一个元素进行结构转换
 *     第二个参数 : 分区内计算规则
 *     第三个参数 : 分区间计算规则
 *
 *
 *
 *
 */
object T28_ByKey_总结 {
  private val log: Log = LogFactory.getLog(T08_RDD使用_从日志中读取uri.getClass)

  def main(args: Array[String]): Unit = {

    //TODO 准备环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)

    //TODO 创建RDD
    val rdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 2), ("b", 3), ("b", 4), ("b", 5), ("a", 6)), 2)

    /**
     *  reduceByKey底层调用       combineByKeyWithClassTag[V]((v: V) => v, func, func, partitioner)
     *  aggregateByKey底层调用    combineByKeyWithClassTag[U]((v: V) => cleanedSeqOp(createZero(), v),
     *
     *   foldByKey底层调用        combineByKeyWithClassTag[V]((v: V) => cleanedFunc(createZero(), v),
     *   combineByKey底层调用     combineByKeyWithClassTag(createCombiner, mergeValue, mergeCombiners, defaultPartitioner(self))
     *
     *
     *   combineByKeyWithClassTag():
     *    第一个参数: 值
     *    第二个参数: 分区内计算规则
     *    第三个参数: 分区间计算规则
     *    第四个参数: 都一样
     *
     */



    rdd.reduceByKey(_ + _)
    rdd.aggregateByKey(0)(_ + _, _ + _)
    rdd.foldByKey(0)(_ + _)
    rdd.combineByKey(v => v, (x: Int, y) => x + y, (x: Int, y: Int) => x + y)


   // resultRDD.collect().foreach(println)


    //TODO 关闭环境
    sc.stop()


  }
}
